package com.springboot.netty.dispatcher;

import com.alibaba.fastjson.JSON;
import com.springboot.netty.codec.Invocation;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.*;

@Component
@ChannelHandler.Sharable
public class MessageDispatcher extends SimpleChannelInboundHandler<Invocation> {
    @Autowired
    private MessageHandlerContainer messageHandlerContainer;

//        private final ExecutorService executor =  Executors.newFixedThreadPool(200);
    private static final int cpu_core = 8;

    private final ExecutorService executor = new ThreadPoolExecutor(cpu_core, cpu_core * 2,
            1L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(500),
            Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy());

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Invocation invocation) throws Exception {
        // 获得 type 对应的 MessageHandler 处理器
        MessageHandler messageHandler = messageHandlerContainer.getMessageHandler(invocation.getType());
        // 获得  MessageHandler 处理器 的消息类
        Class<? extends Message> messageClass = MessageHandlerContainer.getMessageClass(messageHandler);
        // 解析消息
        Message message = JSON.parseObject(invocation.getMessage(), messageClass);
        // 执行逻辑
        executor.submit(new Runnable() {

            @Override
            public void run() {
                // noinspection unchecked
                messageHandler.execute(channelHandlerContext.channel(), message);
            }

        });
    }
}
